package com.example.toy.core.lock;

import com.example.toy.core.locker.ToyLock;
import com.example.toy.core.utils.WaitUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * FileLock 基于HDFS文件的分布式锁
 *
 * @author shenbing
 * @date 2021/12/22
 */
@Slf4j
public class FileLock implements ToyLock {

  private final Path lockFile;
  private final FileSystem fs;
  private long timestamp;

  @SneakyThrows
  public FileLock(Path lockFile) {
    this.lockFile = lockFile;
    this.fs = lockFile.getFileSystem(new Configuration());
    // 创建parent文件夹
    Path parent = lockFile.getParent();
    fs.mkdirs(parent);
  }

  @Override
  public void lock(long time, TimeUnit unit) throws InterruptedException, TimeoutException {
    String msg = "can not create file lock: " + lockFile;
    WaitUtils.wait(
        3_000L,
        unit.convert(time, TimeUnit.MILLISECONDS),
        msg,
        () -> {
          try {
            return fs.createNewFile(lockFile);
          } catch (Exception e) {
            log.warn("Can not create lock file: " + e.getMessage());
            return false;
          }
        });
    try {
      FileStatus fileStatus = fs.getFileStatus(lockFile);
      timestamp = fileStatus.getModificationTime();
    } catch (IOException e) {
      throw new RuntimeException("Can not get file status for lock file: " + lockFile);
    }
  }

  @SneakyThrows
  @Override
  public void unlock() {
    if (locked()) {
      FileStatus fileStatus = fs.getFileStatus(lockFile);
      if (fileStatus.getModificationTime() != this.timestamp) {
        throw new RuntimeException("Can not unlock, the file lock created by this client");
      }
      fs.delete(lockFile, false);
    } else {
      log.warn("unlock a not exists lock");
    }
  }

  @Override
  @SneakyThrows
  public boolean locked() {
    return fs.exists(lockFile);
  }
}
